package com.stu.book.wwj2.char05;

import org.junit.jupiter.api.Test;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Author: MI
 * @Date: 2022/2/19/23:38
 * @Description:
 */
public class CompletableDemo {

    /**
     * 5.5.2 任务的异步运行
     */
    public void sync() {
        // 异步执行Supplier类型的任务
        CompletableFuture<Integer> supplyAsync = CompletableFuture.supplyAsync(() -> 355);

        //异步执行Runnable类型的任务
        CompletableFuture.runAsync(() -> {
            System.out.println();
        });
    }

    /**
     * 异步任务链
     */
    @Test
    public void pipeline() {
        //thenApply：以同步方式继续处理上一个异步任务的结果
        ExecutorService executor = Executors.newFixedThreadPool(3);

        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("supplyAsync:" + Thread.currentThread());
            return "java";
        }, executor).thenApply(e -> {
            System.out.println("thenApply:" + Thread.currentThread());
            return e.length();
        });
        future.join();
    }

    @Test
    public void thenApplyAsync() throws ExecutionException, InterruptedException {
        //thenApplyAsync:以异步的方式继续处理上一个异步运行的结果
        ExecutorService executor = Executors.newFixedThreadPool(3);

        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("supplyAsync:" + Thread.currentThread());
            return "java";
        }, executor).thenApplyAsync(e -> {
            System.out.println("thenApply:" + Thread.currentThread());
            return e.length();
        });
        System.out.println(future.get() == 4);
    }


    /**
     * thenAccept:以同步的方式消费上一个异步任务的结果
     */
    @Test
    public void thenApplyAccept() {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        //  thenAccept:以同步的方式消费上一个异步任务的结果

        CompletableFuture future = CompletableFuture.supplyAsync(() -> {
            System.out.println("supplyAsync:" + Thread.currentThread());
            return "java";
        }, executor).thenAccept(v -> {
            System.out.println("thenApply:" + Thread.currentThread());
            System.out.println(v);
        });
        future.join();
    }

    @Test
    public void thenApplyAcceptAsync() {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        //  thenApplyAcceptAsync:以异步的方式消费上一个异步任务的结果

        CompletableFuture future = CompletableFuture.supplyAsync(() -> {
            System.out.println("supplyAsync:" + Thread.currentThread());
            return "java";
        }, executor).thenAcceptAsync(v -> {
            System.out.println("thenApply:" + Thread.currentThread());
            System.out.println(v);
        });
        future.join();
    }

    /**
     * 合并多个Future
     */
    public void composeAndCombine() {
        CompletableFuture<String> thenCompose = CompletableFuture.supplyAsync(() -> "java")
                // s为上一个Future计算的结果
                .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " scala"));

        CompletableFuture.supplyAsync(() -> "java")
                // s为上一个Future计算的结果
                .thenCombine(CompletableFuture.supplyAsync(() -> " scala"),
                        //s1 s2 分别是第一个和第二个 Future计算的结果
                        (s1, s2) -> s1 + s2);
    }

    public void errHandel() {
        //错误处理
        CompletableFuture.<String>supplyAsync(() -> {
            throw new RuntimeException();
        }).handle((r, e) -> {
            if (e != null) {
                return "error";
            } else {
                return r;
            }
        }).thenAccept(System.out::println);
    }


}
